/*
 * @Author: 来自火星的码农 15122322+heyzhi@user.noreply.gitee.com
 * @Date: 2025-06-30 16:12:29
 * @LastEditors: 来自火星的码农 15122322+heyzhi@user.noreply.gitee.com
 * @LastEditTime: 2025-09-04 07:51:23
 * @FilePath: /server/workspace/src/base/scheduler.cc
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
#include "scheduler.h"
#include "base/Log.h"
#include "base/fiber.h"
#include "base/hook.h"
#include "base/marco.h"
#include "base/mutex.h"
#include "base/thread.h"
#include "base/util.h"
#include <cstddef>
#include <functional>
#include <string>
#include <vector>

namespace Yizhi{

    thread_local Scheduler* t_scheduler=nullptr;
    thread_local Fiber* t_scheduler_fiber=nullptr;
    static Logger::ptr g_logger=SERVER_LOG_NAME("system");
    Scheduler::Scheduler(size_t threads,const std::string &name,bool user_caller)
    
    {   
        m_name=name;
         SERVER_ASSERT(threads>0);
        if(user_caller){
            Yizhi::Fiber::GetThis();
            threads--;
            m_rootFiber.reset(new Fiber(std::bind(&Scheduler::run,this),true));
            SERVER_ASSERT(t_scheduler==nullptr);
            t_scheduler=this;
            t_scheduler_fiber=m_rootFiber.get();
            m_rootThread=Yizhi::GetThreadid();    
            m_threadIds.push_back(m_rootThread);
        }

        else{
            m_rootThread=-1;
        }
        m_threadCount=threads;
    }

    Scheduler::~Scheduler(){
        SERVER_ASSERT(m_stopping);
        if(GetThis()==this){
            t_scheduler=nullptr;
        }
    }


    void Scheduler:: start(){
        MutexType::Lock lock(m_mutex);
        if(!m_stopping){
            return;
        }
        m_stopping=false;

        SERVER_ASSERT(m_threads.empty());
        m_threads.resize(m_threadCount);
        for(int i=0;i<m_threadCount;i++){
            m_threads[i].reset(new Thread(std::bind(&Scheduler::run,this)
            ,m_name+" "+std::to_string(i)));
            m_threadIds.push_back(m_threads[i]->GetTid());
        }
       lock.unlock();

    }


    void Scheduler::stop(){
    
        m_autostop=true;
        //只开当前调度器为协程
        if(m_rootFiber&&
            m_threadCount==0&&
            (m_rootFiber->IsTERM()||m_rootFiber->IsEXCEPT())){
                SERVER_LOG_INFO(g_logger)<<"this stop";
                m_stopping=true;

                if(isStoped()){
                    return;
                }
            }

        if(m_rootThread!=-1){
            SERVER_ASSERT(GetThis()==this);
        }
        else{
            SERVER_ASSERT(GetThis()!=this);
        }
        
        m_stopping=true;

        for(int i=0;i<m_threadCount;i++){
            tickle();
        }

        if(m_rootFiber){
            tickle();
        }
        if(m_rootFiber){
            
            if(!isStoped()){
                m_rootFiber->call();
            }
        }


        std::vector<Thread::ptr> thr;
        {
            MutexType::Lock lock(m_mutex);
            thr.swap(m_threads);
        }
        for(auto i:thr){
            i->join();
        }

    }

    void Scheduler::run(){
        SERVER_LOG_DEBUG(g_logger)<<"run";
        SetThis();
        set_hook_enable(true);
        //假使当前线程并非调度器所在的线程，则需要声明线程上的主协程
        if(GetThreadid()!=m_rootThread){
            t_scheduler_fiber=Fiber::GetThis().get();
        }
        Task task;
        Fiber::ptr idle_fiber(new Fiber(std::bind(&Scheduler::idle,this)));
        Fiber::ptr cb_fiber;

        //开始在循环中取任务，并执行任务
        while (true) {
            task.reset();
            bool tick_me=false;
            bool isactive=false;

            {
                MutexType::Lock lock(m_mutex);
                auto it=m_taskpools.begin();
                while(it!=m_taskpools.end()){
                    if(it->m_threads!=-1&&it->m_threads!=GetThreadid()){
                        it++;
                        tick_me=true;
                        continue;
                    }
                    if(it->m_fiber&&it->m_fiber->IsEXEC()){
                        it++;
                        continue;
                    }
                    task=*it;
                    m_taskpools.erase(it);
                    ++m_activeThreadCount;
                    isactive=true;
                    break;
                }
                tick_me|=it!=m_taskpools.end();
            }

            if(tick_me){
                tickle();
            }
            if(task.m_fiber&&!task.m_fiber->IsTERM()&&!task.m_fiber->IsEXCEPT()){
                task.m_fiber->resunme();
                --m_activeThreadCount;

                ///需要再次执行
                if(task.m_fiber->IsREADY()){
                   scheduler(task.m_fiber); 
                }

                ///非正常结束
                else if(!task.m_fiber->IsTERM()&&!task.m_fiber->IsEXCEPT()){
                    task.m_fiber->setState(Fiber::State::HOLD);
                }
                task.reset();
            }

            else if(task.m_cb){
                ///任务协程为初始化
                if(cb_fiber){
                    cb_fiber->reset(task.m_cb);
                }
                ///协程栈空间复用
                else{
                    cb_fiber.reset(new Fiber(task.m_cb));
                }
                task.reset();
                cb_fiber->resunme();
                --m_activeThreadCount;

                ///需要再次唤醒
                if(cb_fiber->IsREADY()){
                    scheduler(cb_fiber);
                    cb_fiber.reset();
                }
                ///正常结束
                else if(cb_fiber->IsEXCEPT()||cb_fiber->IsTERM()){
                    cb_fiber->reset(nullptr);
                }
                //非正常结束
                else{
                    cb_fiber->setState(Fiber::State::HOLD);
                    cb_fiber.reset();
                }
            }
            else{
                ///其他协程取出任务并执行，当前协程没有任务，需要等待所有任务结束后，和其他线程一起结束
                if(isactive){
                    --m_activeThreadCount;
                    continue;
                }
                if(idle_fiber->IsTERM()){
                    SERVER_LOG_DEBUG(g_logger)<<"idle fiber term";
                    break;
                }
                ++m_idleThreadcount;
                idle_fiber->resunme();
                --m_idleThreadcount;

                if(!idle_fiber->IsEXCEPT()&&!idle_fiber->IsTERM()){
                    idle_fiber->setState(Fiber::State::HOLD);
                }

            }
        }
    }

    void Scheduler::tickle(){
        SERVER_LOG_DEBUG(g_logger)<<"tickle";
    }
    void Scheduler::idle(){
        SERVER_LOG_DEBUG(g_logger)<<"idle";
        while(!isStoped()){
            Fiber::YieldToHold();
        }
    }
    bool Scheduler::isStoped(){
        MutexType::Lock lock(m_mutex);
        return m_stopping&&m_autostop
                &&m_activeThreadCount==00&&m_taskpools.empty();
    }
    void Scheduler::SetThis(){
        t_scheduler=this;
    }

    Fiber* Scheduler::GetMainFiber(){
        return t_scheduler_fiber;
    }

    Scheduler* Scheduler::GetThis(){
        return t_scheduler;
    }

}